package top.hudk.peakclipping;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 作用：利用时间滑动窗口和循环链表实现限流器
 * 可以实现的需求比如：每秒钟最多放行受理10个请求。
 *
 * 时间窗口是一个由两个时间点构成的时间段范围。
 * 时间窗口的长度，就是这段时间段的长短，由调用者决定，默认为一秒钟。
 * 两个时间点暂且分别命名为窗头时间点，和窗尾时间点
 * 窗头时间点：新请求到来时的时间点，也是当下时间点，表示时间窗口的右边界
 * 窗尾时间点：窗头时间点减去时间窗口长度，得到的那个时间点就是窗尾时间点，表示时间窗口的左边界。
 *
 * 循环链表的每个结点与时间窗口内的每次请求一一对应，结点中主要记录请求发生那一刻的时间戳。
 * 循环链表的结点个数代表时间窗口内允许的最大请求个数
 * 举例：(...>7>8>9>)1>2>3>4>5>6>7>8>9>(1>2>3>...)
 * 以上循环链表共有9个结点，因为是循环链表，所以9号结点又会与1号节点连接，逻辑上如同括号中的内容一样向两边延展表示。
 * 定义两个“指针”变量，分别为循环链表的头指针和尾指针，分别对应时间窗口内即将到来的最新请求和最早请求结点
 * 新的请求，永远对应头结点的更新，换句话说：
 * 新的情求到来时，头指针对应的结点将记录此时的时间戳，然后将头指针向右移动，指向新的空结点，为新的请求做准备
 * 尾指针指向的结点，记录着时间窗口内最早的请求时间戳，随着时间的推移，最早的请求对应的时间戳会不断的逐一超出时间窗口的左边界，
 * 故需要不断更新尾结点位置，使尾指针及时的更新到时间窗口左边界之内最左边的结点上。
 *
 * 定义一个【请求是否已满标志】，记录请求数是否达到时间窗口内的最大请求数量限制，
 * 此标志初始为 false，表示时间窗口内的请求数量没有达到最大数量限制
 * 日过请求数达到时间窗口内的最大请求数量限制，代表着头指针比尾指针快了一圈追上了尾指针,此时需要将【请求已满标志】置为 true.
 *
 * 每次新的请求到来时
 * 1、刷新时间窗口：判断尾指针结点的时间戳是否超出了时间窗口左边界，若超过，则尾指针右移一步，然后继续以上判断，
 * 直到尾指针结点的时间戳已经在时间窗口之内，或者尾结点与头结点相遇。
 * （所谓尾指针结点的时间戳超出时间窗口左边界的意思是指：尾指针结点的时间戳 < 窗头时间点（当下时间） - 时间窗口长度）
 * 2、如果【请求已满标志置】为 true，时间窗口内的请求数已经达到最大请求限制，拒绝放行该请求（返回调用者 false）；
 * 如果【请求已满标志置】为 false，表示可以放行接收新的请求。更新头指针结点的时间戳为当下时间点，并将头指针向右移一步，
 * 如果此时头指针与尾指针指向了同一个结点，则说明此时的请求并发量比较密集，时间窗口内的请求数已经达到最大请求限制，此时需要将【请求已满标志】置为 true。
 * 此时当下的请求已经被接收，表示请求被放行（返回调用者 ture）。
 *
 *
 * @author hudk
 * @date 2020/12/16 21:17
 */
public class CircularLinkedCurrentLimiter<E> implements CurrentLimiter{

    /**
     * 默认限流值
     * 时间窗口内最大请求数
     */
    private static final int DEFAULT_CURRENT_LIMIT = 100;

    /**
     * 默认时间窗口长度（单位：毫秒）
     */
    private static final int DEFAULT_TIME_WINDOW = 1000;


    /**
     * 存放所有结点的列表集合
     */
    private final List<Node<E>> nodeList = new ArrayList<>();

    /**
     * 实际限流值
     * 时间窗口内实际允许的最大请求数
     */
    private int currentLimit;

    /**
     * 实际的时间窗口长度（单位：毫秒）
     */
    private int timeWindow;


    /**
     * 时间窗口右边界（窗头时间点）
     */
    private long timeWindowRight;

    /**
     * 时间窗口左边界（窗尾时间点）
     */
    private long timeWindowLeft;

    /**
     * 是否达到最大限流，达到为true，否则为false
     */
    private volatile AtomicBoolean fullFlag = new AtomicBoolean();

    /**
     * 头指针
     */
    private Node<E> head;


    /**
     * 尾指针
     */
    private Node<E> tail;


    /**
     * 实时计数，记录时间窗口内实时请求个数
     */
    private volatile AtomicInteger count = new AtomicInteger();


    /**
     * 打印输出字符串使用
     */
    private StringBuilder stringBuilder = new StringBuilder();


    /**
     * 获取实时流量QPS(平均一秒内的请求次数)
     * @return
     */
    @Override
    public synchronized int getCurrentTimeFlow(){
        //refreshTimeWindow();
        return this.count.get() * 1000 / this.timeWindow;
    }

    /**
     *
     * @param currentLimit 限流值（时间窗口内最大流量）
     * @param timeWindow 时间窗口长度（毫秒）
     */
    CircularLinkedCurrentLimiter(int currentLimit, int timeWindow){
        if(currentLimit < 0){
            throw new RuntimeException("CircularLinkedCurrentLimiter构造器参数不可以小于零：currentLimit。");
        }
        if(timeWindow < 0){
            throw new RuntimeException("CircularLinkedCurrentLimiter构造器参数不可以小于零：timeWindow。");
        }
        this.currentLimit = currentLimit;
        this.timeWindow = timeWindow;
        this.fullFlag.set(false);
        //初始化循环单链表
        initialization(currentLimit);
    }

    /**
     *
     * @param currentLimit 限流值（时间窗口内最大流量）
     */
    CircularLinkedCurrentLimiter(int currentLimit){
        this(currentLimit, DEFAULT_TIME_WINDOW);
    }

    CircularLinkedCurrentLimiter(){
        this(DEFAULT_CURRENT_LIMIT, DEFAULT_TIME_WINDOW);
    }

    /**
     * 初始化循环单链表
     * @param linkedLength 链表长度
     */
    private void initialization(int linkedLength){
        for(int i = 0; i < linkedLength; i++){
            Node<E> node = createNode();
            node.number = i + 1;
            nodeList.add(node);
            if(i > 0){
                nodeList.get(i-1).next = node;
            }
            if(i == linkedLength - 1 ){
                node.next = nodeList.get(0);
            }
        }
        this.head = nodeList.get(0);
        this.tail = nodeList.get(0);
    }

    /**
     * 时间窗口计数，
     * 当实时流量大于限流值时，返回false,
     * 小于等于限流值时返回true
     * @return
     */
    @Override
    public synchronized boolean count(){
        //刷新时间窗口
        refreshTimeWindow();
        //移动尾结点至时间窗口内
        moveTailPointerToTimeWindow();
        //如果达到流量限值，返回false
        if (this.fullFlag.get()){
            return false;
        }
        //没有达到限流值，处理以下逻辑
        //更新头指针结点的时间戳
        this.head.timeStamp = this.timeWindowRight;
        //头指针步进,指向下一个结点
        //如果步进后头指针指向的结点就是尾部结点，说明达到限流值
        if(Objects.equals((this.head = this.head.next), this.tail)){
            this.fullFlag.set(true);
        }
        //流量值加1
        count.incrementAndGet();
        return true;
    }

    /**
     * 刷新时间窗口
     */
    private void refreshTimeWindow(){
        this.timeWindowRight = getCurrentTimeMillis();
        this.timeWindowLeft = this.timeWindowRight - this.timeWindow;
    }

    /**
     * 移动尾指针直至到达时间窗口内
     */
    private void moveTailPointerToTimeWindow(){
        //初始态时，不需要刷新时间窗口，如果流量为0也不做刷新
        if(tail.timeStamp == 0 || count.get() == 0){
            return;
        }
        // 从尾端开始顺着链表遍历，如果节点的时间戳已经超出时间窗口范围，尾指针向前进一步
        // 直到尾结点与头结点相遇，或者如果遇到时间戳没有超出时间窗口的节点，则跳出遍历
        do {
            //从尾指针向头指针方向，离头结点越近，结点的时间戳越晚，也就是离当下时间越近。
            //离尾结点越近，结点的时间戳越早，也就是离当下时间越远，越先超出时间窗口
            //所以，如果遇到时间戳没有超出时间窗口的节点，说明下一个结点的时间戳肯定也没有超出时间窗口，可以跳出遍历
            if (tail.timeStamp >= this.timeWindowLeft) {
                break;
            }
            //尾指针向前进一步
            tail = tail.next;
            count.decrementAndGet();
            //将限流值满的标志置为false
            fullFlag.set(false);
        } while (!Objects.equals(tail, head));
    }

    /**
     * 获取当下的毫秒时间戳值
     * @return 毫秒时间戳值
     */
    private long getCurrentTimeMillis(){
        return System.currentTimeMillis();
    }

    /**
     * 打印循环链表双指针移动状态
     * @return
     */
    @Override
    public synchronized String toString(){
        //refreshTimeWindow();
        StringBuilder sb = stringBuilder;
        sb.delete(0,sb.length());
        // 结点是否被请求占用标识
        // 如果头结点在尾结点右边，则位于尾结点右边且位于头结点左边的节点，是被请求占用的节点，flag初始为false
        // 如果头结点在尾结点左边，则位于尾结点左边或位于头结点右边的节点，是被请求占用的节点，flag初始为true
        // 在未达到限流值得情况下，头结点和尾结点两则的状态肯定不一样，所以遇到头结点或尾结点，需要将flag翻转
        // 若达到限流值，说明头结点和尾结点重合，且其他全部结点都是被占用状态
        boolean flag = this.head == this.tail ? this.fullFlag.get() : this.tail.number > this.head.number;
        for(int i = 0; i < nodeList.size(); i++){
            Node<E> node = nodeList.get(i);
            //如果头结点和尾结点重合了
            if(node == this.head && node == this.tail){
                sb.append("●");
                continue;
            }
            //遇到头结点，将flag翻转
            if(node == this.head){
                sb.append(">");
                flag = !flag;
                continue;
            }
            //遇到尾结点，将flag翻转
            if(node == this.tail){
                sb.append("●");
                flag = !flag;
                continue;
            }
            if(flag){
                sb.append("●");
            }else {
                sb.append("◌");
            }
        }
        sb.append("\t[" + (this.fullFlag.get()?"●":"◌")  + "]");
        //sb.append("[" + this.count.get() + "]");
        sb.append("[" + getCurrentTimeFlow() + "]");
        sb.append("\r");
        return sb.toString();
    }

    /**
     * 输出实时流量柱状图
     * @return
     */
    public synchronized String toString2(){
        //refreshTimeWindow();
        StringBuilder sb = stringBuilder;
        sb.delete(0,sb.length());
        boolean flag = false;
        Node<E> node = this.tail;
        do {
            //遇到头结点，将flag翻转
            if(node == this.head){
                flag = !flag;
            }
            //遇到尾结点，将flag翻转
            if(node == this.tail){
                flag = !flag;
            }
            if(this.fullFlag.get() || flag){
                sb.append("●");
            }else {
                sb.append("◌");
            }
            node = node.next;
        } while (!Objects.equals(node, this.tail));
        sb.append("\t[" + (this.fullFlag.get()?"●":"◌")  + "]");
        //sb.append("[" + this.count.get() + "]");
        sb.append("[" + getCurrentTimeFlow() + "]");
        sb.append("\r");
        return sb.toString();
    }

    /**
     * 创建一个新的结点
     * @return
     */
    private Node<E> createNode(){
        return new Node<E>();
    }


    /**
     * 单链表结点对象
     * @param <E>
     */
    private static class Node<E>{

        /**
         * 时间戳
         */
        volatile long timeStamp;

        /**
         * 元素对象
         */
        E element;

        /**
         * 下一个结点引用
         */
        Node<E> next;

        /**
         * 结点编号，没啥用，这个属性可以没有
         */
        int number;

        Node(){}

        Node(long timeStamp,E e){
            this.timeStamp = timeStamp;
            this.element = e;
        }

        public final long getTimeStamp()        { return timeStamp; }
        public final E getElement()      { return element; }
        @Override
        public final String toString() { return String.valueOf(number); }

        @Override
        public final int hashCode() {
            return Objects.hashCode(timeStamp) ^ Objects.hashCode(element);
        }

        public final E setElement(E element) {
            E oldElement = element;
            this.element = element;
            return oldElement;
        }

        @Override
        public final boolean equals(Object o) {
            if (o == this){
                return true;
            }
            return false;
        }

    }
}
